package com.rocketmq.comsumer;

import com.rocketmq.Client;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.Collections;

public class MyPushConsumer extends Client {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    public void consumer(String group, String topic) throws ClientException {
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        FilterExpression filterExpression = new FilterExpression();
        provider.newPushConsumerBuilder()
                //设置消费者分组。
                .setConsumerGroup(group)
                //设置接入点。
                .setClientConfiguration(getConfig())
                //设置预绑定的订阅关系。
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                //设置消费监听器。
                .setMessageListener(new MessageListener() {
                    @Override
                    public ConsumeResult consume(MessageView messageView) {
                        String value = StandardCharsets.UTF_8.decode(messageView.getBody()).toString();
                        logger.info("消息内容:{}", value);
                        //消费消息并返回处理结果。
                        return ConsumeResult.SUCCESS;
                    }
                }).build();
        // pushConsumer.close();
    }
}